package com.example.messageproducer;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.Test;

import javax.jms.*;
import java.io.IOException;

public class ConsumerWithTransaction {

    private static final String BROKER_URL = "tcp://localhost:61616";
    private static final String QUEUE_NAME = "queue2";

    @Test
    public void testConsumerFromQueueWithTransaction() throws JMSException, IOException {

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
        Connection connection = connectionFactory.createConnection();

        // transacted用于指示要创建的Session是否是事务性的。
        // 当transacted为true的时候，表示Session是事务性的，此时createSession方法的第二个参数会被忽略
        final boolean transacted = true;
        Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);

        Queue queue = session.createQueue(QUEUE_NAME);

        MessageConsumer messageConsumer = session.createConsumer(queue);

        messageConsumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
               TextMessage textMessage = (TextMessage) message;
                try {
                    System.out.println("收到消息：" + textMessage.getText());

                    // 提交事务
                    session.commit();
                } catch (JMSException e) {
                    e.printStackTrace();

                    // 回滚事务
                    try {
                        session.rollback();
                    } catch (JMSException jmsException) {
                        jmsException.printStackTrace();
                    }
                }
            }
        });

        connection.start();


        System.in.read();
        session.close();
        connection.close();
    }
}
